package com.example.demo.kafka;

import com.alibaba.fastjson.JSON;
import com.example.demo.util.LogUtils;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.sql.Timestamp;

/**
 * @author zhangxinju@hellobike.com
 * @date 2019-08-22 14:15
 **/
@Component
@EnableScheduling
public class KafkaProducer implements InitializingBean {

    private final KafkaTemplate<String,String> kafkaTemplate;

    @Autowired
    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(cron = "00/30 * * * * ?")
    private void send(){
        Student student = new Student(System.currentTimeMillis(), "Jack", 12);
        LogUtils.COMMON.info("student:{}", JSON.toJSONString(student));
        kafkaTemplate.send("student", JSON.toJSONString(student));
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        send();
    }

    @Setter
    @Getter
    @AllArgsConstructor
    private static class Student {
        private long timestamp;
        private String username;
        private Integer age;
    }
}

